package com.hujf.flink.DataStreamAPI;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @Author hujf
 * @Date 2022/3/24 14:51
 * @Description TODO
 */
public class WindowJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.getConfig().setAutoWatermarkInterval(15L);
        DataStreamSource<Integer> integerDataStreamSource = environment.fromElements(1, 2, 3);
        DataStreamSource<Integer> integerDataStreamSource2 = environment.fromElements(1, 4, 6);
        DataStream<Tuple2<Integer, Integer>> apply = integerDataStreamSource.join(integerDataStreamSource2)
                .where(data -> data)
                .equalTo(data -> data)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new JoinFunction<Integer, Integer, Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> join(Integer integer, Integer integer2) throws Exception {
                        return new Tuple2<>(integer, integer2);
                    }
                });
        apply.print();
        environment.execute("TimeWindowDemo");


//
//        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
//        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//        SingleOutputStreamOperator<Tuple2<String,Integer>> source = environment.fromElements(
//                Tuple2.of("a",1),
//                Tuple2.of("b",2),
//                Tuple2.of("d",4)
//                ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String,Integer>>forMonotonousTimestamps().withTimestampAssigner(
//                new SerializableTimestampAssigner<Tuple2<String,Integer>>() {
//                    @Override
//                    public long extractTimestamp(Tuple2<String, Integer> stringIntegerTuple2, long l) {
//                        return stringIntegerTuple2.f1;
//                    }
//                }
//        ));
//        SingleOutputStreamOperator<Tuple2<String,Integer>> source1 = environment.fromElements(
//                Tuple2.of("a",1),
//                Tuple2.of("b",2),
//                Tuple2.of("e",3)
//        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String,Integer>>forMonotonousTimestamps().withTimestampAssigner(
//                new SerializableTimestampAssigner<Tuple2<String,Integer>>() {
//                    @Override
//                    public long extractTimestamp(Tuple2<String, Integer> stringIntegerTuple2, long l) {
//                        return stringIntegerTuple2.f1;
//                    }
//                }
//        ));
//        DataStream<Tuple3<String, String, Integer>> apply = source.join(source1)
//                .where(data -> data.f0)
//                .equalTo(data -> data.f0)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, String, Integer>>() {
//
//                    @Override
//                    public Tuple3<String, String, Integer> join(Tuple2<String, Integer> a1, Tuple2<String, Integer> a2) throws Exception {
//                        return new Tuple3<>(a1.f0, a2.f0, a1.f1 + a2.f1);
//                    }
//                });
//        apply.print();
//        environment.execute();
    }


}
